package com.yanggu.spark.core.rdd.transform.value

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

//RDD转换算子 map 和 mapPartitions
//这两个算子并没有没有本质上的区别。如果有外部链接操作等, 使用mapPartitions算子可以减少类似的操作。
//同时mapPartitions算子是将一个分区内的数据全部发送给计算节点，容易造成OOM，使用时需要预估分区的数据量大小。
//RDD根据数据处理的方式将算子整体上可以划分为Value类型、双Value类型、Key-Value类型
object RDD04_Map_Map_Partitions_Transform_Operator {

  def main(args: Array[String]): Unit = {
    //1. 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("mapOperator")

    //2. 创建SparkContext
    val sparkContext = new SparkContext(sparkConf)

    //3. 创建RDD算子
    //两个分区1, 2, 3 & 4, 5, 6, 7
    val dataRdd: RDD[Int] = sparkContext.makeRDD(1 to 7, 2)

    //4. map算子
    //map转换算子和scala的map类似。
    //将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。
    val dataRdd2: RDD[Int] = dataRdd.map{
      println("map方法执行了。。。。。。")
      _ * 2
    }

    println("****分区数" + dataRdd2.getNumPartitions)

    //5. mapPartitions算子
    //mapPartitions算子是将一个分区内的数据全部一起执行。如果在映射的过程中存在，外部资源。例如JDBC的连接等, hbase的连接等。使用mapPartitions算子的效率更高。这样不用频繁地创建连接。
    //mapPartitions算子也有一些问题。是将一个分区的数据发送到计算节点进行处理。如果数据量过大容易造成OOM
    val value = dataRdd2.mapPartitions {
      iter => {
        println("获取外部连接。。。")
        val iterator = iter.map(_ * 2)
        println("释放外部连接")
        iterator
      }
    }
    println(value.collect().mkString(", "))

    //6. mapPartitionsWithIndex
    //和mapPartitions类似，不过这里带上了分区索引
    val value1 = dataRdd2.mapPartitionsWithIndex {
      (index, datas) => {
        println("mapPartitionsWithIndex方法执行了")
        datas.map {
          data => (index, data * 2)
        }
      }
    }
    println(value1.collect().mkString(", "))

    //7. 打印
    dataRdd2.collect().foreach(println)

    //6. 释放资源
    sparkContext.stop()
  }

}
